package com.yaorange.teaching.service.impl;

import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import com.alibaba.druid.sql.ast.expr.SQLCaseExpr;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.yaorange.common.core.exception.ServiceException;
import com.yaorange.common.core.utils.DateUtils;
import com.yaorange.common.core.utils.IKAnalyzerUtils;
import com.yaorange.teaching.bo.HighInterviewMasteryDetailBo;
import com.yaorange.teaching.bo.QueryInterviewMasteryDegreeBo;
import com.yaorange.teaching.bo.RelatedInterviewBo;
import com.yaorange.teaching.index.HighInterviewIndex;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.QueryBuilders;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;
import com.yaorange.teaching.mapper.YaHighInterviewMapper;
import com.yaorange.teaching.domain.YaHighInterview;
import com.yaorange.teaching.service.IYaHighInterviewService;
import org.springframework.transaction.annotation.Transactional;

/**
 * 高频面试题Service业务层处理
 *
 * @author ruoyi
 * @date 2024-11-15
 */
@Slf4j
@Service
public class YaHighInterviewServiceImpl implements IYaHighInterviewService
{
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private YaHighInterviewMapper yaHighInterviewMapper;

    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    //定义一个单线程池
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>());

    /**
     * 查询高频面试题
     *
     * @param id 高频面试题主键
     * @return 高频面试题
     */
    @Override
    public YaHighInterview selectYaHighInterviewById(Long id)
    {
        return yaHighInterviewMapper.selectYaHighInterviewById(id);
    }

    /**
     * 查询高频面试题列表
     *
     * @param yaHighInterview 高频面试题
     * @return 高频面试题
     */
    @Override
    public List<YaHighInterview> selectYaHighInterviewList(YaHighInterview yaHighInterview)
    {
        return yaHighInterviewMapper.selectYaHighInterviewList(yaHighInterview);
    }

    /**
     * 新增高频面试题
     *
     * @param yaHighInterview 高频面试题
     * @return 结果
     */
    @Override
    public int insertYaHighInterview(YaHighInterview yaHighInterview)
    {
        yaHighInterview.setCreateTime(DateUtils.getNowDate());
        //修改mysql成功后
        int i = yaHighInterviewMapper.insertYaHighInterview(yaHighInterview);
        try {
            //发消息通知即可(业务解耦,异步处理)
            rabbitTemplate.convertAndSend("edu_smart.interview.exchange", "interview.add", yaHighInterview);
        } catch (Exception e) {
            log.error("发送到mq失败", e);
        }
        /*//1.同步es

        //2.同步redis*/
        return 1;
    }

    /**
     * 修改高频面试题
     *
     * @param yaHighInterview 高频面试题
     * @return 结果
     */
    @Override
    public int updateYaHighInterview(YaHighInterview yaHighInterview)
    {
        yaHighInterview.setUpdateTime(DateUtils.getNowDate());
        return yaHighInterviewMapper.updateYaHighInterview(yaHighInterview);
    }

    /**
     * 批量删除高频面试题
     *
     * @param ids 需要删除的高频面试题主键
     * @return 结果
     */
    @Override
    public int deleteYaHighInterviewByIds(Long[] ids)
    {
        return yaHighInterviewMapper.deleteYaHighInterviewByIds(ids);
    }

    /**
     * 删除高频面试题信息
     *
     * @param id 高频面试题主键
     * @return 结果
     */
    @Override
    public int deleteYaHighInterviewById(Long id)
    {
        return yaHighInterviewMapper.deleteYaHighInterviewById(id);
    }

    @Override
    public List<HighInterviewMasteryDetailBo> selectHighInterviewMasteryDetailList(QueryInterviewMasteryDegreeBo queryInterviewMasteryDegreeBo) {
        return yaHighInterviewMapper.selectHighInterviewMasteryDetailList(queryInterviewMasteryDegreeBo);
    }

    @Override
    @Transactional(readOnly = true)
    public RelatedInterviewBo selectRelatedInterview(Long cdId, String courseContent) {
        // 课程内容解析成 关键词列表，比如 "我是中国人"   我,中国，中国人，国人（使用中文分词器  ik分词器）
        Set<String> keywords = IKAnalyzerUtils.parseSentence(courseContent, true);
        if (keywords.isEmpty()) {
            throw new ServiceException("没有解析出对应的分词");
        }
        //先查询es,es不可用再查mysql
        List<YaHighInterview> interviewList = null;

        try {
            NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
            //分词查询
            nativeSearchQueryBuilder.withQuery(QueryBuilders.matchQuery("question", courseContent));
            SearchHits<HighInterviewIndex> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), HighInterviewIndex.class);

            //将HighInterviewIndex -> YaHighInterview
            interviewList = searchHits.getSearchHits().stream().map(searchHit -> {
               HighInterviewIndex content = searchHit.getContent();
               return new YaHighInterview(content.getId(), content.getStage(), content.getQuestion(), content.getAnswer(), content.getKnowledgePoint(), content.getImportance());
           }).collect(Collectors.toList());
        } catch (Exception e) {
            // 根据关键词列表去数据库模糊查询 高频面试题
            interviewList = yaHighInterviewMapper.selectYaHighInterviewListByKeywords(keywords);
        }

        // 根据课程id查询出已经关联过的面试题id列表
        List<Long> relatedInterviewIdList = yaHighInterviewMapper.selectRelatedInterviewIdList(cdId);
        return new RelatedInterviewBo(interviewList,keywords,relatedInterviewIdList);
    }

    private volatile int percentage = 0;

    //定义 一个信号量(1)
//    private final Semaphore semaphore = new Semaphore(1);
    @Autowired
    private RedissonClient redissonClient;
    @Override
    @Transactional(readOnly = true)
    public void exportEs() {
        //做一个并发限制，判断是否有线程正在做全量同步，如果有人在做就直接返回，并提示
        RSemaphore semaphore = redissonClient.getSemaphore("exportEs");
        semaphore.trySetPermits(1);
        if(!semaphore.tryAcquire()){
            throw new ServiceException("已经有全量同步任务在执行，请稍后再试");
        }
        THREAD_POOL_EXECUTOR.submit(()->{

            try {
                rebuildIndex();
                //分批次读取
                int pageSize=1000;
                long size=0;
                int pageNo = 1;
                
                int saveCount = 0;
                do{
                    log.debug("正在同步第{}页数据",pageNo);
                    //查询一页数据
                    QueryWrapper<YaHighInterview> queryWrapper = new QueryWrapper<>();
                    Page<YaHighInterview> page = yaHighInterviewMapper.selectPage(new Page<>(pageNo++, pageSize), queryWrapper);
                    size = page.getRecords().size();
                    saveCount+= (int) size;
                    log.debug("当前页数据量:{},累计:{}", size, saveCount);
                    //数据转换，将YaHighInterview->HighInterviewIndex
                    List<HighInterviewIndex> highInterviewIndices =
                            page.getRecords().stream()
                                    .map(item ->
                                         new HighInterviewIndex(item.getId(), item.getStage(), item.getQuestion(), item.getAnswer(), item.getKnowledgePoint(), item.getImportance())
                                    ).collect(Collectors.toList());
                    //将这一页数据同步到es
                    elasticsearchRestTemplate.save(highInterviewIndices);

                    percentage = (int) ((double) saveCount /(page.getTotal()*0.01));
                    log.debug("正在同步第{}页数据结束,进度:{}%",pageNo-1,percentage);
                }while (size==pageSize);
            } catch (Exception e) {
                log.error("同步数据失败",e);
            } finally {
                //解锁(防止死锁)
                semaphore.release();
            }
        });

    }

    @Override
    public int queryPercentage() {
        return percentage;
    }

    private void rebuildIndex() {
        //重建索引库
        IndexOperations indexOperations = elasticsearchRestTemplate.indexOps(HighInterviewIndex.class);
        indexOperations.delete();
        // 创建索引，会根据Item类的@Document注解信息来创建
        indexOperations.create();
        // 配置映射，会根据Item类中的id、Field等字段来自动完成映射
        indexOperations.putMapping(indexOperations.createMapping(HighInterviewIndex.class));
    }
}
